#ifndef __rpc_udp_bkjl6sdoi7fjwe90e3wf9__
#define __rpc_udp_bkjl6sdoi7fjwe90e3wf9__

#include <string>
#include <memory>
#include <thread>

#include <hude/base/block_queue.hpp>
#include <hude/aio/network.hpp>
#include <hude/utils/parses.hpp>
#include <hude/utils/json.hpp>
#include <hude/rpc_json/rpc_json.hpp>
#include <hude/rpc_json/filters/filter_default.hpp>
#include <hude/base/hex.hpp>
#include <hude/safe/crc.hpp>

namespace hude
{
    namespace rpc
    {
        template< typename _channel, std::size_t _count, typename _filter >
        struct _rpc_channel_t :
            public std::enable_shared_from_this< _rpc_channel_t< _channel, _count, _filter > >
        {
            //-------------------------------------------------------------
            using this_t        = _rpc_channel_t< _channel, _count, _filter >;
            using net_t         = hude::net::net_t< _count >;
            using filter_t      = _filter;
            using rpc_t         = typename hude::rpc::rpc_t< this_t >;
            using parses_t      = hude::utils::parses_t< uint8_t >;
            using checker_t     = utils::_details::json_checker_t< typename parses_t::citr_t >;


            //-------------------------------------------------------------
            typename net_t::ctx_t       aio_;
            std::weak_ptr< _channel >   chl_;
            typename rpc_t::context_t   ctx_;
            filter_t                    filter_;
            parses_t                    parses_;
            checker_t                   checker_;
            std::thread                 thread_;

            /*********************************************************************
             * @brief _rpc_channel_t
             */
            _rpc_channel_t( typename net_t::ctx_t aio, std::shared_ptr< _channel > chl ) : aio_( aio ), chl_( chl )
            {
                // log_dug( "\t rpc_channel_t( %p )::rpc_channel_t( %p )", this, aio.get() );
            }

            /*********************************************************************
             * @brief ~rpc_channel_t
             */
            ~_rpc_channel_t()
            {
                // log_dug( "\t rpc_channel_t::~rpc_channel_t( %p )", this );
                parses_.observer_pre( parses_t::observer_t{} );
                parses_.cannel();
                if( thread_.joinable() ) thread_.join();
            }

            /*********************************************************************
             * @brief 构建RPC服务
             */
            bool service( std::function< void( typename rpc_t::context_t ) > cb_entry )
            {
                // 初始化RPC上下文
                ctx_ = rpc_t::make_context( this->shared_from_this() );

                // 登记RPC服务
                cb_entry( ctx_ );

                // 桢查找器, 从祼数据中扫描一个桢结构的方法
                auto scaner = [=]( parses_t::citr_t& b, parses_t::citr_t& e ) -> parses_t::citr_t
                {
                    static const checker_t checker;

                    parses_t::citr_t h = e;

                    auto ret = checker( h, b, e, true );

                    return ret? h : e;
                };

                // 无效数据丢弃事件
                auto droping = []( parses_t::citr_t b, parses_t::citr_t e ) -> void
                {
                    log_war( "droped bytes: %u(%s)", e - b, hude::utils::to_hex( &(*b), e - b ).c_str() );
                };

                auto cb_push = [=]( std::string frame )
                {
                    // 数据过滤及处理
                    filter_.recv( frame );
                    ctx_->on_recv( frame );
                };

                // 数据桢的处理回调
                auto cb_frame = [=]( parses_t::citr_t& b, parses_t::citr_t& e, bool& ) -> bool
                {
                    // log_dug( "\t rpc_net_t::aio_( %p )", aio_.get() );
                    aio_->task_commit( std::bind( cb_push, std::string{ b, e } ) );

                    // cb_push( std::string{ b, e } );

                    return true;
                };

                // 设置数据桢的处理
                parses_.observer_pre( cb_frame );

                // 开始解析栈循环
                thread_ = std::thread( [=]{ parses_.process( scaner, droping ); } );

                return true;
            }

            /*********************************************************************
             * @brief 发送数据
             */
            intptr_t send( std::string& frame )
            {
                // log_dug( "\t channel( %p ) send  >>>  %u: %s", this, frame.size(), frame.c_str() );

                if( chl_.expired() )
                {
                    return -1000;
                }

                filter_.send( frame );

                auto ptr = chl_.lock();

                // log_dug( "\t\t send buffer{ %u: %s }", buffer.size(), hude::utils::to_hex( buffer ).c_str() );

                auto ret = ptr->send( &(frame[0]), frame.size() );

                auto siz = (int)frame.size();

                return ( ret > siz)? siz : ret;
            }

            /*********************************************************************
             * @brief 数据接收事件的处理回调
             */
            void on_read( std::shared_ptr< _channel > chl )
            {
                int ret = 0;

                do
                {
                    // 数据接收缓存
                    auto fragment = parses_.buffer_alloc();

                    // 接收数据，返回值-1为没有数据可接收，0为网络断开了
                    ret = chl->recv( *fragment );

                    if( ret > 0 )
                    {
                        parses_.buffer_push( fragment );
                    }
                } while ( ret > 0 );

                // log_dug( "\t channel( %p ) recv  <<<  %u: %s", this, fragment->size(), &((*fragment)[0]) );
            }
  
            /*********************************************************************
             * @brief 网络关闭事件的处理回调
             */
            void on_close( std::shared_ptr< _channel > chl )
            {
                // log_nte( "rpc channel( %p ) closed.", this );

                parses_.observer_pre( parses_t::observer_t{} );
                parses_.cannel();
            }
        };

        /*********************************************************************
         * @brief RPC加载器基础
         */
        template< typename _service >
        struct _rpc_listener_t
        {
            using types_t  = hude::net::net_traits_t< _service >;
            using net_t    = hude::net::net_t< types_t::emCOUNT >;

            static typename std::shared_ptr< typename types_t::type_t > io_obj(
                std::function< void( typename types_t::ptr_channel_t ) > cb, typename net_t::ctx_t ctx, const typename net_t::address_t& addr )
            {
                auto cb_accepted = [=]( const typename net_t::address_t& host, typename types_t::ptr_channel_t chl )
                {
                    log_nte( "accepted client{ host: %s }, chl( %p )", host.str().c_str(), chl.get() );
                    cb( chl );
                };

                return net_t::template io_obj< typename types_t::type_t >( ctx, addr, cb_accepted );
            }
        };

        /*********************************************************************
         * @brief RPC加载器基础
         */
        template< typename _service >
        struct _rpc_connector_t
        {
            using types_t  = hude::net::net_traits_t< _service >;
            using net_t    = hude::net::net_t< types_t::emCOUNT >;

            static typename std::shared_ptr< typename types_t::type_t > io_obj(
                std::function< void( typename types_t::ptr_channel_t ) > cb, typename net_t::ctx_t ctx, const typename net_t::address_t& addr )
            {
                return net_t::template io_obj< typename types_t::type_t >( ctx, addr, cb );
            }
        };

        /*********************************************************************
         * @brief RPC加载器基础
         */
        template< typename _service, std::size_t _count = 0U >
        struct _rpc_boot_t {};

        template< std::size_t _count >
        struct _rpc_boot_t< hude::net::tcp_listen_t< _count > > : public _rpc_listener_t< hude::net::tcp_listen_t< _count > > {};

        template< std::size_t _count >
        struct _rpc_boot_t< hude::net::tcp_connect_t< _count > > : public _rpc_connector_t< hude::net::tcp_connect_t< _count > > {};

        template< std::size_t _count >
        struct _rpc_boot_t< hude::net::udp_vlink_t< _count > > : public _rpc_listener_t< hude::net::udp_vlink_t< _count > > {};

        template< std::size_t _count >
        struct _rpc_boot_t< hude::net::udp_connect_t< _count > > : public _rpc_connector_t< hude::net::udp_connect_t< _count > > {};


        /*********************************************************************
         * @brief RPC加载器基础
         */
        template< typename _service, typename _filter >
        struct _rpc_loader_t
        {
            //-------------------------------------------------------------
            using this_t         = _rpc_loader_t< _service, _filter >;
            using types_t        = hude::net::net_traits_t< _service >;
            using net_t          = hude::net::net_t< types_t::emCOUNT >;
            using rpc_channel_t  = _rpc_channel_t< typename types_t::raw_channel_t, types_t::emCOUNT, _filter >;
            using ptr_channel_t  = std::shared_ptr< rpc_channel_t >;
            using rpc_t          = typename hude::rpc::rpc_t< rpc_channel_t >;


            //-------------------------------------------------------------
            typename net_t::ctx_t       ctx_;
            typename net_t::service_t   service_;

            /*********************************************************************
             * @brief 默认构造
             */
            _rpc_loader_t() : ctx_( net_t::context_t::ctx() )
            {
            }

            /*********************************************************************
             * @brief 外部提供网络上下文的构造
             */
            _rpc_loader_t( typename net_t::ctx_t ctx ) : ctx_( ctx )
            {
            }

            /*********************************************************************
             * @brief 数据接收事件的处理回调
             */
            void cancel()
            {
                ctx_.cancel();
            }

            /*********************************************************************
             * @brief 启动加载服务
             */
            bool boot( const typename net_t::address_t& source, std::function< void( typename rpc_t::context_t ) > cb_rpc_entry,
                std::function< void() > cb_close = std::function< void() >{} )
            {
                // 连接请求事件处理
                auto cb = [=]( typename types_t::ptr_channel_t chl )
                {
                    // 为新客户链接创建RPC服务实例
                    chl->user_.template emplace< ptr_channel_t >( new rpc_channel_t{ chl->ctx_, chl } );

                    hude::any_cast< ptr_channel_t >( chl->user_ )->service( cb_rpc_entry );

                    // 绑定数据接收事件到RPC实例
                    chl->on_recv_  = [=]( typename types_t::ptr_channel_t chl ) { hude::any_cast< ptr_channel_t >( chl->user_ )->on_read ( chl ); };
                    // 绑定网络关闭事件到RPC实例
                    chl->on_close_ = [=]( typename types_t::ptr_channel_t chl ) { hude::any_cast< ptr_channel_t >( chl->user_ )->on_close( chl ); };
                };
        
                // 端口监听服务的入口
                {
                    // 服务端开始监听端口
                    auto service = _rpc_boot_t< _service >::io_obj( cb, ctx_, source );

                    service_ = std::dynamic_pointer_cast< typename net_t::service_t::element_type >( service );

                    // 监听端口的关闭事件
                    service_->on_close_ = [=]( typename net_t::sock_t sock ) { if( cb_close ) cb_close(); };
        
                    // 监听及失败处理
                    auto ret = service_->state();

                    if( ! ret )
                    {
                        log_err( "socket error: %s", service_->error() );
                        service_->close();
                    }

                    return ret;
                }
            }

            /*********************************************************************
             * @brief 启动加载服务
             */
            bool boot( const char* addr, uint16_t port, std::function< void( typename rpc_t::context_t ) > cb_rpc_entry,
                std::function< void() > cb_close = std::function< void() >{} )
            {
                auto source = typename net_t::address_t{ addr, port };
                return boot( source, cb_rpc_entry, cb_close );
            }

            /*********************************************************************
             * @brief 启动加载服务并运行事件泵服务
             */
            int run( const typename net_t::address_t& source, std::function< void( typename rpc_t::context_t ) > cb_rpc_entry )
            {
                auto ret = ctx_.run( [=]() { this->boot( source, cb_rpc_entry ); } );

                // 退出状态提示
                if( ret < 0 )
                {
                    log_err( "ctx error( %d ): %s", ret, ctx_.error() );
                }

                return ret;
            }

            /*********************************************************************
             * @brief 启动加载服务并运行事件泵服务
             */
            int run( const char* addr, uint16_t port, std::function< void( typename rpc_t::context_t ) > cb_rpc_entry )
            {
                return run( typename net_t::address_t{ addr, port }, cb_rpc_entry );
            }
        };

        /*********************************************************************
         * @brief _rpc_booter_connector_t
         */

        /*********************************************************************
         * @brief rpc_loader_t
         */
        template< typename _service, typename _filter = filter_default_t >
        using rpc_loader_t = _rpc_loader_t< _service, _filter >;

    } //namespace rpc
} //namespace hude

#endif //__server_udp_bkjl6sdoi7fjwe90e3wf9__
